package com.atguigu.flink.chapter08_exec2;

import com.atguigu.flink.pojo.AdsClickLog;
import com.atguigu.flink.pojo.LoginEvent;
import com.atguigu.flink.pojo.MyUtil;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.windowing.windows.GlobalWindow;
import org.apache.flink.util.Collector;

import java.util.List;

/**
 * Created by Smexy on 2022/10/29
 *
 * 如果同一用户（可以是不同IP）在2秒之内连续两次登录失败，就认为存在恶意登录。
 *
 *
 *
 *      使用基于元素个数的窗口。 
 *          前提，数据必须是有序的。
 *          假设数据有序。
 *
 */
public class Demo5_LoginFail
{
    public static void main(String[] args) {
        
        
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        env.setParallelism(1);

        env.readTextFile("data/LoginLog.csv").setParallelism(1)
           .map(new MapFunction<String, LoginEvent>()
           {
               @Override
               public LoginEvent map(String value) throws Exception {
                   String[] words = value.split(",");
                   return new LoginEvent(
                       Long.parseLong(words[0]),
                       words[1],
                       words[2],
                       Long.parseLong(words[3]) * 1000
                   );
               }
           })
           .keyBy(LoginEvent::getUserId)
           .countWindow(2,1)
           .process(new ProcessWindowFunction<LoginEvent, String, Long, GlobalWindow>()
           {
               @Override
               public void process(Long uid, Context context, Iterable<LoginEvent> elements, Collector<String> out) throws Exception {

                   List<LoginEvent> loginEvents = MyUtil.toList(elements);

                   //只有窗口中有两次登录信息，才需要去判断
                   if (loginEvents.size() == 2){

                       LoginEvent e1 = loginEvents.get(0);
                       LoginEvent e2 = loginEvents.get(1);

                       if ("fail".equals(e1.getEventType())
                            &&
                           "fail".equals(e2.getEventType())
                           &&
                           Math.abs(e1.getEventTime() - e2.getEventTime()) < 2000
                       ){
                           out.collect(uid + "  涉嫌恶意登录");

                       }
                   }

               }
           }).print().setParallelism(1);
        
        try {
                    env.execute();
                } catch (Exception e) {
                    e.printStackTrace();
                }
    }
}
